package com.mall.common.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.function.Consumer;

/**
 * RocketMQ工具类
 * 通过new创建实例，支持多例模式
 */
@Slf4j
public class RocketMqUtil {
    
    private String nameServer;
    private String groupName;
    private DefaultMQProducer producer;
    private DefaultMQPushConsumer consumer;
    
    /**
     * 构造函数
     * 
     * @param nameServer NameServer地址
     * @param groupName  生产者/消费者组名
     */
    public RocketMqUtil(String nameServer, String groupName) {
        this.nameServer = nameServer;
        this.groupName = groupName;
    }
    
    /**
     * 初始化生产者
     * 
     * @throws MQClientException MQ客户端异常
     */
    public void initProducer() throws MQClientException {
        if (producer == null) {
            producer = new DefaultMQProducer(groupName);
            producer.setNamesrvAddr(nameServer);
            producer.start();
            log.info("RocketMQ生产者初始化成功，NameServer: {}, GroupName: {}", nameServer, groupName);
        }
    }
    
    /**
     * 初始化消费者
     * 
     * @param topic      主题
     * @param tag        标签
     * @param listener   消息监听器
     * @throws MQClientException MQ客户端异常
     */
    public void initConsumer(String topic, String tag, MessageListenerConcurrently listener) throws MQClientException {
        if (consumer == null) {
            consumer = new DefaultMQPushConsumer(groupName);
            consumer.setNamesrvAddr(nameServer);
            consumer.subscribe(topic, tag);
            consumer.registerMessageListener(listener);
            consumer.start();
            log.info("RocketMQ消费者初始化成功，NameServer: {}, GroupName: {}, Topic: {}, Tag: {}", 
                    nameServer, groupName, topic, tag);
        }
    }
    
    /**
     * 发送同步消息
     * 
     * @param topic   主题
     * @param tag     标签
     * @param body    消息内容
     * @return SendResult 发送结果
     * @throws Exception 异常
     */
    public SendResult sendSyncMessage(String topic, String tag, String body) throws Exception {
        if (producer == null) {
            throw new IllegalStateException("生产者未初始化");
        }
        
        Message msg = new Message(topic, tag, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        return producer.send(msg);
    }
    
    /**
     * 发送异步消息
     * 
     * @param topic    主题
     * @param tag      标签
     * @param body     消息内容
     * @param callback 回调函数
     * @throws Exception 异常
     */
    public void sendAsyncMessage(String topic, String tag, String body, SendCallback callback) throws Exception {
        if (producer == null) {
            throw new IllegalStateException("生产者未初始化");
        }
        
        Message msg = new Message(topic, tag, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.send(msg, callback);
    }
    
    /**
     * 发送单向消息
     * 
     * @param topic 主题
     * @param tag   标签
     * @param body  消息内容
     * @throws Exception 异常
     */
    public void sendOnewayMessage(String topic, String tag, String body) throws Exception {
        if (producer == null) {
            throw new IllegalStateException("生产者未初始化");
        }
        
        Message msg = new Message(topic, tag, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.sendOneway(msg);
    }
    
    /**
     * 注册消费者监听器
     * 
     * @param messageHandler 消息处理器
     * @return 消息监听器
     */
    public MessageListenerConcurrently createMessageListener(Consumer<String> messageHandler) {
        return new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String body = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        log.info("接收到消息: {}", body);
                        messageHandler.accept(body);
                    } catch (UnsupportedEncodingException e) {
                        log.error("消息编码异常: ", e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    } catch (Exception e) {
                        log.error("消息处理异常: ", e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        };
    }
    
    /**
     * 关闭生产者
     */
    public void shutdownProducer() {
        if (producer != null) {
            producer.shutdown();
            log.info("RocketMQ生产者已关闭");
        }
    }
    
    /**
     * 关闭消费者
     */
    public void shutdownConsumer() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("RocketMQ消费者已关闭");
        }
    }
    
    /**
     * 关闭所有资源
     */
    public void shutdown() {
        shutdownProducer();
        shutdownConsumer();
    }
    
    // Getters and Setters
    public String getNameServer() {
        return nameServer;
    }
    
    public void setNameServer(String nameServer) {
        this.nameServer = nameServer;
    }
    
    public String getGroupName() {
        return groupName;
    }
    
    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }
    
    public DefaultMQProducer getProducer() {
        return producer;
    }
    
    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }
}